package org.example.es;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.TrustStrategy;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.StopWatch;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.junit.Before;
import org.junit.Test;

import javax.net.ssl.SSLContext;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;

@Slf4j
public class ElasticSearchTest {

    private static final String userName = "admin";

    private static final String password = "admin";

    private RestHighLevelClient client;

    private BulkProcessor bulkProcessor;

    private Map<Long, StopWatch> map = new ConcurrentHashMap<>();

    private static final AtomicInteger counter = new AtomicInteger(0);

    @Before
    public void before() throws Exception {

        HttpHost[] httpHosts = new HttpHost[1];
        httpHosts[0] = new HttpHost("192.168.206.122", 9200, "https");
//        httpHosts[0] = new HttpHost("localhost", 9200, "http");

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY.ANY, new UsernamePasswordCredentials(userName, password));

        SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
            // 信任所有
            public boolean isTrusted(X509Certificate[] chain, String authType) throws CertificateException {
                return true;
            }
        }).build();
        SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE);

        RestClientBuilder builder = RestClient.builder(httpHosts).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                requestConfigBuilder.setConnectTimeout(-1);
                requestConfigBuilder.setSocketTimeout(-1);
                requestConfigBuilder.setConnectionRequestTimeout(-1);
                return requestConfigBuilder;
            }
        }).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                httpClientBuilder.disableAuthCaching();
                 httpClientBuilder.setSSLStrategy(sessionStrategy);
                 return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
//                return httpClientBuilder;
            }
        });

        client = new RestHighLevelClient(builder);

        // 构建BulkProcessor
        BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
                (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);

        bulkProcessor = BulkProcessor.builder(bulkConsumer, new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                //在这儿你可以自定义执行同步之前执行什么
                StopWatch stopWatch = new StopWatch();
                stopWatch.start("task--" + executionId);
                map.put(executionId, stopWatch);
            }

            @SneakyThrows
            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                //在这儿你可以自定义执行完同步之后执行什么
                // log.info("{}写入成功", executionId);
                StopWatch stopWatch = map.get(executionId);
                if (stopWatch != null) {
                    stopWatch.stop();
                    log.info("executionId={}, {}，导入完成：{}条", executionId, stopWatch.shortSummary(), request.numberOfActions());
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                //写入失败后
                log.error("ES写入失败", failure);
                failure.printStackTrace();
            }
        })
                .setBulkActions(10000) //  达到刷新的条数
                // .setFlushInterval(TimeValue.timeValueSeconds(120)) // 固定刷新的时间频率
                .setFlushInterval(TimeValue.timeValueSeconds(-1)) // 固定刷新的时间频率
                .setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB))
                // 0代表同步提交即只能提交一个request；
                // 1代表当有一个新的bulk正在累积时，1个并发请求可被允许执行
                // 并发线程数
                .setConcurrentRequests(10)
                .build();
    }

    @Test
    public void testExists() throws Exception {

        GetRequest getRequest = new GetRequest(
                "posts",
                "1");
        getRequest.fetchSourceContext(new FetchSourceContext(false));
        getRequest.storedFields("_none_");

        log.info("posts existed = {}", client.exists(getRequest, RequestOptions.DEFAULT));
    }

    /**
     * 测试批量插入索引
     * bulk 是 不限制数量的
     *
     * @throws Exception
     */
    @Test
    public void testBulk() throws Exception {

        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.timeout("1000s");

        List<TagList> tagList = new ArrayList<>();
        tagList.add(new TagList(1l, "性别", "男性"));
        for (int i = 0; i < tagList.size(); i++) {
            //批量更新删除
            TagList tagList1 = tagList.get(i);
            Map<String, Object> map = new HashMap<>();
            map.put("tag_id", tagList1.getTagId());
            map.put("tag_name", tagList1.getTagName());
            map.put("tag_value", tagList1.getTagValue());
            bulkRequest.add(new IndexRequest("ep_tag_001").id("ep_tag_001").source(map));
        }
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        log.info("result = {}", bulkResponse.hasFailures());
    }

    @Test
    public void deleteIndex() throws Exception {
        DeleteIndexRequest request = new DeleteIndexRequest("ep_tag_001");
        client.indices().delete(request,RequestOptions.DEFAULT);
        log.info("delete success");
    }

    /**
     * bulkProcessor是可以显示数量的，避免Bulk一下子太多导致ES挂掉
     *
     * @throws Exception
     */
    @Test
    public void testBulkProcessor() throws Exception {
//        List<Map<String, Object>> list = initTagList();
//        for (Map<String, Object> map : list) {
//            bulkProcessor.add(new IndexRequest("ep_tag_001").id("ep_tag_001").source(map));
//        }

        while (true) {
            Map<String, Object> map = new HashMap<>();
            int tagId = counter.getAndIncrement();
            map.put("tag_id", tagId);
            map.put("tag_name", "name" + tagId);
            map.put("tag_value", "value" + tagId);
            // 记录的ID不要设置，设置之后每次ES都要检查一遍
            bulkProcessor.add(new IndexRequest("ep_tag_001").source(map));
        }

        // Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    public void after() throws Exception {
        client.close();
    }

    public List<Map<String, Object>> initTagList() {
        List<TagList> tagList = new ArrayList<>();
        tagList.add(new TagList(1l, "性别", "男性"));
        
        List<Map<String, Object>> list = new ArrayList<>();
        for (TagList tagList1 : tagList) {
            Map<String, Object> map = new HashMap<>();
            map.put("tag_id", tagList1.getTagId());
            map.put("tag_name", tagList1.getTagName());
            map.put("tag_value", tagList1.getTagValue());
            list.add(map);
        }
        return list;
    }

    @Test
    public void testPrint() {
        log.info("testPrint...");
    }
}
